package com.atguigu.flink.demo04.chapter_6;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.junit.Test;

/**
 * 页面访问数，不去重
 * @author admin
 * @date 2021/8/10
 */
public class PvProject {


    @Test
    public void test1() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        
        DataStreamSource<String> source = env.readTextFile("D:\\project\\idea\\flink\\input\\UserBehavior.csv");

        SingleOutputStreamOperator<Tuple2<String, Integer>> pv1 = source.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            String[] split = value.split(",");
            String pv = split[3];
            if ("pv".equals(pv)) {
                out.collect(Tuple2.of("pv", 1));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.INT));

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = pv1.keyBy(s -> s.f0).sum(1);

        sum.print("sum>>>");


        env.execute();

    }

    @Test
    public void test2() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> source = env.readTextFile("D:\\project\\idea\\flink\\input\\UserBehavior.csv");

        SingleOutputStreamOperator<Tuple2<String, Integer>> pv1 = source.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            String[] split = value.split(",");
            String pv = split[3];
            if ("pv".equals(pv)) {
                out.collect(Tuple2.of("pv", 1));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.INT));


        SingleOutputStreamOperator<Tuple2<String, Integer>> process = pv1.process(new ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            private int count = 0;

            @Override
            public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                count += value.f1;
                out.collect(Tuple2.of(value.f0, count));
            }
        });

        process.print().setParallelism(1);
        

      //  SingleOutputStreamOperator<Tuple2<String, Integer>> sum = process.keyBy(s -> s.f0).sum(1);

      //  sum.print("sum>>>");


        env.execute();

    }


}
